package com.atguigu.flink.wordcount;

import com.atguigu.flink.pojo.Wordcount;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author WEIYUNHUI
 * @date 2023/6/10 11:15
 *
 * 使用POJO来封装数据.
 *    flink对POJO的要求:
 *       1. 类必须是public
 *       2. 类中必须有无参数构造器
 *       3. 类中的属性要么是public的， 要么是private的，但是必须要提供public 的get/set方法
 *       4. 类中的属性必须都能序列化.
 *
 */
public class Flink06_POJOWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1) ;

        // 2.读取数据
        // DataStreamSource => DataStream
        DataStreamSource<String> ds = env.socketTextStream("127.0.0.1", 8888) ;

        // 3.转换处理
        // 3.1 切分数据，处理成 Wordcount
        SingleOutputStreamOperator<Wordcount> flatMapDs = ds.flatMap(
                        //lambda写法
                        (String line, Collector<Wordcount> out) -> {
                            String[] words = line.split(" ");
                            for (String word : words) {
                                out.collect(new Wordcount(word, 1));
                            }
                        }
                )//.returns(Wordcount.class) ;
                .returns(Types.POJO(Wordcount.class));
        // 3.2 按照单词分组
        KeyedStream<Wordcount, String> keyByDs = flatMapDs.keyBy(
                // wordcount -> wordcount.getWord()
                Wordcount::getWord
        );
        // 3.3 汇总
        // 使用Tuple中的第二个元素进行汇总
        SingleOutputStreamOperator<Wordcount> sumDs = keyByDs.sum("count");

        // 4. 输出结果
        sumDs.print();

        // 5. 启动执行
        env.execute();
    }

}


















